package com.zfei.firstflink.utils;

import com.zfei.firstflink.model.XexHomeOrder;
import org.apache.commons.math3.util.Decimal64;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import scala.math.Numeric;

import java.math.BigDecimal;
import java.sql.*;
import java.util.Timer;
import java.util.TimerTask;


/***
 *$com.zfei.firstflink.utils.MysqlSource
 *@author:jiwenfei
 *
 *@version:
 *XexHomeOrder
 *createTime:2018/8/17:上午10:15
 *
 *description:
 */
public class MysqlSource2 extends RichSourceFunction<XexHomeOrder> {
    Connection connection=null;
    PreparedStatement ps=null;
    int total=0;
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        String driver="com.mysql.jdbc.Driver";
        String url="jdbc:mysql://47.100.10.9:3306/civil_admin";
//        String url="jdbc:mysql://192.168.1.151:3306/civil_administration";
        String userName="zkxbxmysql";
        String passward="Zkxbx-2011!";
//        String passward="zkxbx2011"
        Class.forName(driver);
        connection=DriverManager.getConnection(url,userName,passward);


        String sql="select * from xex_home_order  order by order_time asc limit 1,10000";
        ps=connection.prepareStatement(sql);
    }

    @Override
    public void run(SourceContext<XexHomeOrder> ctx) throws Exception {
        try {
            int start=0;
            int page=total/500;


            ResultSet rs= ps.executeQuery();
            while (rs.next()){
                XexHomeOrder order=new XexHomeOrder();
                try{
                    String tmporderNo=rs.getString("order_no");
//                   if (tmporderNo==null||tmporderNo.trim().equals("")){
//                       order.setOrderNo("werrrrrrr");
//                   }else {
//                       order.setOrderNo(new String(tmporderNo.getBytes("GBK"),"UTF-8"));
//                   }
                    order.setOrderNo(tmporderNo);

                    //System.out.println(order.getOrderNo());
                    order.setCustomerId(rs.getString("customer_id"));

                    order.setServiceObjectId(rs.getNString("service_object_id"));
                    order.setServiceProviderId(rs.getNString("service_provider_id"));
                    order.setServicePersonId(rs.getNString("service_person_id"));
                    order.setState(rs.getInt("state"));
                    // System.out.println(order.getState());
                    order.setOrderType(rs.getInt("order_type"));
                    order.setNumber(rs.getNString("number"));
                    String price=rs.getString("price");
                    if (price==null||"".equals(price)||"NaN".equals(price)){
                        order.setPrice(new BigDecimal("0"));
                    }else {
                        try {
                            order.setPrice(new BigDecimal(price));
                        }catch (Exception e){
                            e.printStackTrace();
                        }
                    }

                     price=rs.getString("spotprice");
                    if (price==null||"".equals(price)||"NaN".equals(price)){
                        order.setSpotprice(new BigDecimal("0"));
                    }else {
                        try {
                            order.setSpotprice(new BigDecimal(price));
                        }catch (Exception e){
                            e.printStackTrace();
                        }
                    }

                    String tmpTime=rs.getString("order_time");
                    try {
                        if ("0000-00-00 00:00:00".equals(tmpTime)){
                            order.setOrderTime(new Timestamp(System.currentTimeMillis()));
                            continue;
                        }else {
                            order.setOrderTime(rs.getTimestamp("order_time"));

                        }
                    }catch (Exception ex){

                        ex.printStackTrace();
                        continue;
                    }
//                   tmpTime=rs.getString("business_time");
//                   if ("0000-00-00 00:00:00".equals(tmpTime)){
//                       order.setBusinessTime(new Date());
//                   }else {
//                       order.setBusinessTime(rs.getDate("business_time"));
//                   }
                    order.setServiceTime(rs.getString("service_time"));
//

                }catch ( Exception c){
                    c.printStackTrace();
                }
                ctx.collectWithTimestamp(order,order.getOrderTime().getTime());
//
//                Timer timer = new Timer();
//                timer.schedule(new TimerTask() {
//                    public void run() {
//
//                    }
//                }, 500);// 设定指定的时间time,此处为2000毫秒

            }
        }catch (Exception ex){
            ex.printStackTrace();
        }
    }

    @Override
    public void cancel() {

    }

    @Override
    public void close() throws Exception {
        super.close();
        if (connection!=null){
            connection.close();
        }
        if (ps!=null){
            ps.close();
        }
    }
}
